package cn.doitedu.rtmk.demo2;


import cn.doitedu.rtmk.demo1.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

/**
 *
 *
 * 实时监控app上的所有用户的所有行为
 * 相较demo1的变化： 规则中，对目标受众，添加了画像约束
 * 规则 1： 当 画像标签 age>=30 and age<=40 AND gender=male  用户发生了 x 行为，立刻推出消息
 * 规则 2： 当 画像标签 active_level=3  AND gender=female 用户发生了 c 行为，且行为属性中符合  properties[p1] = v1
 *
测试数据：
 {"uid":1,"eventId":"x","timestamp":1692844887400,"properties":{"p1":"v4"}}
 {"uid":3,"eventId":"c","timestamp":1692844887400,"properties":{"p1":"v1"}}
 {"uid":2,"eventId":"x","timestamp":1692844887400,"properties":{"p1":"v1"}}
 {"uid":3,"eventId":"x","timestamp":1692844887400,"properties":{"p1":"v1"}}
 */
public class Demo2 {

    public static void main(String[] args) throws Exception {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 开启checkpoint
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setCheckpointTimeout(2000);
        // 设置状态的backend
        env.setStateBackend(new HashMapStateBackend());


        // 构建 kafka source，读取用户实时行为数据
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setGroupId("doit40-1")
                .setTopics("dwd_events")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 读取kafka中的数据为一个流
        DataStreamSource<String> eventsStr = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");

        // 解析行为日志为javabean
        SingleOutputStreamOperator<cn.doitedu.rtmk.demo1.EventBean> beanStream = eventsStr.map(json -> JSON.parseObject(json, cn.doitedu.rtmk.demo1.EventBean.class));


        // 把相同用户的行为，发到相同的subtask去处理
        KeyedStream<cn.doitedu.rtmk.demo1.EventBean, Long> keyedStream = beanStream.keyBy(EventBean::getUid);


        // 规则处理核心逻辑
        SingleOutputStreamOperator<String> resultStream = keyedStream.process(new KeyedProcessFunction<Long, EventBean, String>() {

            Table userProfileTable;
            JSONObject jsonObject;

            @Override
            public void open(Configuration parameters) throws Exception {

                org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
                conf.set("hbase.zookeeper.quorum", "doitedu:2181");

                Connection hbaseConnection = ConnectionFactory.createConnection(conf);
                userProfileTable = hbaseConnection.getTable(TableName.valueOf("user_profile"));

                jsonObject = new JSONObject();
            }

            @Override
            public void processElement(EventBean eventBean, KeyedProcessFunction<Long, EventBean, String>.Context context, Collector<String> collector) throws Exception {

                /**
                 * 规则1
                 */
                // 判断事件 bean中的 eventId=要求的x
                if (eventBean.getEventId().equals("x")) {

                    // 先根据事件 bean中的 uid，去画像库查询该用户的标签是否满足规则要求的条件

                    // 指定查询条件的行健
                    Get getParam = new Get(Bytes.toBytes(eventBean.getUid()));
                    // 指定要查询的标签（qualifier）
                    getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("age"));
                    getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("gender"));


                    // 调用客户端查询
                    Result result = userProfileTable.get(getParam);
                    byte[] ageBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("age"));
                    byte[] genderBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("gender"));
                    int age = Integer.parseInt(Bytes.toString(ageBytes));
                    String genderStr = Bytes.toString(genderBytes);

                    // 判断该用户的画像标签值，是否满足规则要求的值
                    if (age >= 30 && age <= 40 && "male".equals(genderStr)) {

                        jsonObject.put("uid", eventBean.getUid());
                        jsonObject.put("timestamp", eventBean.getTimestamp());
                        jsonObject.put("rule_id", "rule-001");

                        collector.collect(jsonObject.toJSONString());
                    }

                }


                /**
                 * 规则2
                 */
                // 判断事件 bean中的 eventId=要求的x
                if (eventBean.getEventId().equals("c") && eventBean.getProperties().getOrDefault("p1", "").equals("v1")) {

                    // 先根据事件 bean中的 uid，去画像库查询该用户的标签是否满足规则要求的条件

                    // 指定查询条件的行健
                    Get getParam = new Get(Bytes.toBytes(eventBean.getUid()));
                    // 指定要查询的标签（qualifier）
                    getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("active_level"));
                    getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("gender"));


                    // 调用客户端查询
                    Result result = userProfileTable.get(getParam);
                    byte[] activeLevelBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("active_level"));
                    byte[] genderBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("gender"));
                    int activeLevel = Integer.parseInt(Bytes.toString(activeLevelBytes));
                    String genderStr = Bytes.toString(genderBytes);


                    // 判断该用户的画像标签值，是否满足规则要求的值
                    if (activeLevel == 3 && "female".equals(genderStr)) {

                        jsonObject.put("uid", eventBean.getUid());
                        jsonObject.put("timestamp", eventBean.getTimestamp());
                        jsonObject.put("rule_id", "rule-002");

                        collector.collect(jsonObject.toJSONString());

                    }

                }


            }
        });


        resultStream.print();

        env.execute();


    }


}
